package org.example.state;

import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

import java.time.Duration;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/24 16:51
 */
public class OperatorListStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        environment.setParallelism(2);

        SingleOutputStreamOperator<Long> streamOperator = environment
                .socketTextStream("127.0.0.1", 7777)
                .map(new MyMapFunction());

        streamOperator
                .print();

        environment.execute();
    }

    //  1.实现 CheckpointedFunction 接口
    public static class MyMapFunction implements MapFunction<String, Long>, CheckpointedFunction {

        private Long count = 0L;

        private ListState<Long> listState;

        @Override
        public Long map(String value) throws Exception {
            return ++count;
        }

        /**
         * 本地变量拷贝到算子状态中，开启Checkpointed才会调用
         */
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("snapshotState....");
            listState.clear();
            listState.add(count);
        }

        /**
         * 初始化本地变量： 程序启动时恢复时，从算子状态中恢复数据到本地变量中，每个子任务调用一次
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("initializeState....");
            // 从 上下文 初始化 算子状态
            this.listState = context.getOperatorStateStore()
                    .getListState(new ListStateDescriptor<>("listState", Types.LONG));

            // UnionList
//            this.listState = context.getOperatorStateStore()
//                    .getUnionListState(new ListStateDescriptor<>("listState", Types.LONG));

            if (context.isRestored()) {
                // 从 算子状态中 把数据 拷贝到 本地变量
                for (Long aLong : this.listState.get()) {
                    System.out.println("aLong: " + aLong);
                    count += aLong;
                }
            }
        }
    }

    // UnionList:
}
